package flinkstudy.application;


import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.List;
import java.util.Map;
import java.util.Properties;

public class FlinkCEPJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "test");

        DataStream<TemperatureEvent> inputEventStream = env.addSource(
                new FlinkKafkaConsumer<TemperatureEvent>("temperature", new EventDeserializationSchema(), properties));

        /***
         * 定义过滤条件
         */
        Pattern<TemperatureEvent, ?> warningPattern = Pattern.<TemperatureEvent> begin("first")
                .subtype(TemperatureEvent.class).where(
                   new IterativeCondition<TemperatureEvent>() {

                       @Override
                       public boolean filter(TemperatureEvent temperatureEvent, Context<TemperatureEvent> context) throws Exception {
                           if (temperatureEvent.getTemperature() >= 26.0) {
                               return true;
                           }
                           return false;
                       }
                    }
                ).within(Time.seconds(10));


        /**
         * 对过滤后的条件进行结果汇总
         *
         */
        DataStream<Alert> patternStream = CEP.pattern(inputEventStream, warningPattern)
                .select(new PatternSelectFunction<TemperatureEvent, Alert>() {
                    @Override
                    public Alert select(Map<String, List<TemperatureEvent>> event) throws Exception {
                        return new Alert("Temperature Rise Detected:" + event.get("first").size()+" times");
                    }
                });

        patternStream.print();
        env.execute("CEP on Temperature Sensor");


    }
}
